package com.atguigu.gmall.realtime.tutorial.window;


import com.atguigu.gmall.realtime.tutorial.beans.SensorReading;
import com.atguigu.gmall.realtime.tutorial.beans.WindowBean;
import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;

/**
 * @ClassName: WindowTest1_TimeWindow
 * @Description:
 * @Author: wushengran on 2020/11/9 14:37
 * @Version: 1.0
 */
public class WindowTest1_TimeWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

//        // 从文件读取数据
        DataStream<String> inputStream = env.readTextFile("K:\\Project\\flink_project\\flink_real_dataware\\gmall0821realtime\\src\\main\\resources\\sensor_filesystem.txt");

        // socket文本流
//        DataStream<String> inputStream = env.socketTextStream("localhost", 7777);

        // 转换成SensorReading类型
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        SingleOutputStreamOperator<SensorReading> sensorReadingSingleOutputStreamOperator = dataStream.assignTimestampsAndWatermarks(WatermarkStrategy.<SensorReading>forBoundedOutOfOrderness(Duration.ofSeconds(1)).withTimestampAssigner(
                (SerializableTimestampAssigner<SensorReading>) (sensorReading, l) -> sensorReading.getTimestamp()
        ));


        // 开窗测试

//        // 1. 增量聚合函数
//        DataStream<Integer> resultStream = dataStream.keyBy("id")
////                .countWindow(10, 2);
////                .window(EventTimeSessionWindows.withGap(Time.minutes(1)));
////                .window(TumblingProcessingTimeWindows.of(Time.seconds(15)))
//                .timeWindow(Time.seconds(15))
//                .aggregate(new AggregateFunction<SensorReading, Integer, Integer>() {
//                    @Override
//                    public Integer createAccumulator() {
//                        return 0;
//                    }
//
//                    @Override
//                    public Integer add(SensorReading value, Integer accumulator) {
//                        return accumulator + 1;
//                    }
//
//                    @Override
//                    public Integer getResult(Integer accumulator) {
//                        return accumulator;
//                    }
//
//                    @Override
//                    public Integer merge(Integer a, Integer b) {
//                        return a + b;
//                    }
//                });

        // 2. 全窗口函数
        SingleOutputStreamOperator<WindowBean> resultStream2 =
                sensorReadingSingleOutputStreamOperator
//                        .keyBy("id")
//                        .keyBy((KeySelector<SensorReading, String>) SensorReading::getId)
                        .keyBy(SensorReading::getId)
                .timeWindow(Time.seconds(15))
                        .reduce(new ReduceFunction<SensorReading>() {
                            @Override
                            public SensorReading reduce(SensorReading sensorReading, SensorReading t1) throws Exception {
                                sensorReading.setTemperature(sensorReading.getTemperature()+ t1.getTemperature());
                                return sensorReading;
                            }
                        },
//                .process(new ProcessWindowFunction<SensorReading, Object, Tuple, TimeWindow>() {
//                })
//                .apply(new WindowFunction<SensorReading, Tuple3<String, Long, Integer>, Tuple, TimeWindow>() {
//                    @Override
//                    public void apply(Tuple tuple, TimeWindow window, Iterable<SensorReading> input, Collector<Tuple3<String, Long, Integer>> out) throws Exception {
//                        String id = tuple.getField(0);
//                        Long windowEnd = window.getEnd();
//                        Integer count = IteratorUtils.toList(input.iterator()).size();
//                        out.collect(new WindowBean());
//                    }
//                });
        new ProcessWindowFunction<SensorReading, WindowBean, String, TimeWindow>() {
            @Override
            public void process(String tuple, Context context, Iterable<SensorReading> elements, Collector<WindowBean> out) throws Exception {
                SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd hh:MM:ss");
                int count =0;
                for(SensorReading sensorReading : elements){
                    count +=1;
                    String startDate = new DateTime(context.window().getStart(), DateTimeZone.forID("+08:00")).toString("yyyy-MM-dd HH:mm:ss");
                    String endDate = new DateTime(context.window().getEnd(), DateTimeZone.forID("+08:00")).toString("yyyy-MM-dd HH:mm:ss");
                    WindowBean windowBean = new WindowBean(sensorReading.getId(),sensorReading.getTimestamp(),sensorReading.getTemperature(),startDate,endDate);
                    out.collect(windowBean);
                }
                System.out.println(elements.iterator().next().getId() + ">>>>>>>" + count);
            }
        });


        resultStream2.print(">>>>   ");

//        // 3. 其它可选API
//        OutputTag<SensorReading> outputTag = new OutputTag<SensorReading>("late") {
//        };
//
//        SingleOutputStreamOperator<SensorReading> sumStream = dataStream.keyBy("id")
//                .timeWindow(Time.seconds(15))
////                .trigger()
////                .evictor()
//                .allowedLateness(Time.minutes(1))
//                .sideOutputLateData(outputTag)
//                .sum("temperature");

//        sumStream.getSideOutput(outputTag).print("late");

//        resultStream2.print();


        env.execute();
    }
}

